package com.zidongxiangxi.sharding.job.core.node;

import com.zidongxiangxi.sharding.job.core.ShardingJobConfig;
import com.zidongxiangxi.sharding.job.core.storage.NodeStorage;
import com.zidongxiangxi.sharding.job.core.util.BlockUtils;
import lombok.extern.slf4j.Slf4j;

import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;

/**
 * 分片节点
 *
 * @author chenxudong
 * @date 2021/02/18
 */
@Slf4j
public class ShardingNode extends AbstractNode {
    private static final String ROOT = "sharding";
    private static final String PROCESSING = ROOT + "/processing";
    private static final String ITEMS = ROOT + "/items";
    private static final String SHARDING_ITEM = ITEMS + "/%s";
    private static final String SHARDING_ITEM_INSTANCE = SHARDING_ITEM + "/instance";
    private static final String SHARDING_ITEM_RUNNING = SHARDING_ITEM + "/running";

    private final String processingFullPath = getFullPath(PROCESSING);

    public ShardingNode(final String jobName, final NodeStorage storage) {
        super(jobName, storage);
    }
    
    public void setShardingProcessing() {
        storage.fillEphemeralJobNode(PROCESSING, "");
    }

    public void resetShardingInfo(final List<String> items) {
        List<String> oldItems = storage.getJobNodeChildrenKeys(ITEMS);
        if (Objects.nonNull(oldItems)) {
            for (String oldItem : oldItems) {
                storage.removeJobNodeIfExisted(getShardingItemPath(oldItem));
            }
        }
        for (String item : items) {
            storage.createJobNodeIfNeeded(getShardingItemPath(item));
        }
    }

    public void setShardingItemRunning(String shardingItem) {
        storage.fillEphemeralJobNode(getShardingItemRunningPath(shardingItem), "");
    }

    public void waitingOtherShardingItemCompleted(ShardingJobConfig config) {
        while (hasRunningItems(config)) {
            log.debug("任务【{}】短暂休眠，等待分片任务完成", jobName);
            BlockUtils.waitingShortTime();
        }
    }

    private boolean hasRunningItems(ShardingJobConfig config) {
        for (String shardingItem : config.getShardingItems()) {
            if (storage.isJobNodeExisted(getShardingItemRunningPath(shardingItem))) {
                return true;
            }
        }
        return false;
    }

    public Map<String, List<String>> sharding(final List<String> instanceIds, final List<String> totalShardingItems) {
        Map<String, List<String>> shardingResult = shardingAliquot(instanceIds, totalShardingItems);
        addAliquant(instanceIds, totalShardingItems, shardingResult);
        return shardingResult;
    }

    private Map<String, List<String>> shardingAliquot(final List<String> instanceIds,
        final List<String> totalShardingItems) {
        Map<String, List<String>> result = new LinkedHashMap<>(instanceIds.size(), 1);
        int itemCountPerSharding = totalShardingItems.size() / instanceIds.size();
        int count = 0;
        for (String each : instanceIds) {
            List<String> shardingItems = new ArrayList<>(itemCountPerSharding + 1);
            for (int i = count * itemCountPerSharding; i < (count + 1) * itemCountPerSharding; i++) {
                shardingItems.add(totalShardingItems.get(i));
            }
            result.put(each, shardingItems);
            count++;
        }
        return result;
    }

    private void addAliquant(final List<String> instanceIds, final List<String> totalShardingItems, final Map<String,
        List<String>> shardingResults) {
        int shardingTotalCount = totalShardingItems.size();
        int aliquant = shardingTotalCount % instanceIds.size();
        int count = 0;
        for (Map.Entry<String, List<String>> entry : shardingResults.entrySet()) {
            if (count < aliquant) {
                entry.getValue().add(totalShardingItems.get(shardingTotalCount - aliquant + count));
            }
            count++;
        }
    }

    public void changeShardingInstance(String item, String instanceId) {
        storage.fillJobNode(getShardingItemInstancePath(item), instanceId);
    }

    public void clearShardingItemRunning(List<String> shardingItems) {
        for (String shardingItem : shardingItems) {
            storage.removeJobNodeIfExisted(getShardingItemRunningPath(shardingItem));
        }
    }

    public void clearShardingProcessing() {
        storage.removeJobNodeIfExisted(PROCESSING);
    }

    public String getProcessingFullPath() {
        return processingFullPath;
    }

    public boolean isProcessingPath(final String path) {
        return Objects.equals(processingFullPath, path);
    }

    public boolean isShardingItemInstancePath(final String path) {
        return path.startsWith(getFullPath(ITEMS)) && path.endsWith("/instance");
    }

    /**
     * 获取分片节点路径
     *
     * @param item 分片
     * @return 节点路径
     */
    public String getShardingItemPath(final String item) {
        return String.format(SHARDING_ITEM, item);
    }

    public String getShardingItemInstanceId(final String item) {
        return storage.getJobNodeData(getShardingItemInstancePath(item));
    }

    public String getShardingItemInstancePath(final String item) {
        return String.format(SHARDING_ITEM_INSTANCE, item);
    }

    public String getShardingItemRunningPath(final String item) {
        return String.format(SHARDING_ITEM_RUNNING, item);
    }
}
